package com.lewis.sync;

import com.lewis.sync.entity.ColumnInfo;
import com.lewis.sync.entity.ConnInfo;
import com.lewis.sync.entity.TableInfo;
import lombok.extern.slf4j.Slf4j;

import java.sql.*;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.TimerTask;

/**
 * 程序主要定时器
 *
 * @author Lewis
 */
@Slf4j
public class SyncTask extends TimerTask {

	// 编写需要同步的表的行数据（两年的表）
	public static void main(String[] args) {
		int year = 2021;
		int month = 1;
		StringBuilder sb = new StringBuilder();
		for (int i = 1; i <= 24; i++) {
			sb.append("is_year_");
			sb.append(year);
			sb.append(month);
			sb.append(",id,cniStatus,0,100;");
			sb.append("bdbsd_");
			sb.append(year);
			sb.append(month);
			sb.append(",id,cniStatus,0,100;");
			for (int j = 1; j < 7; j++) {
				sb.append("bdbsd_");
				sb.append(year);
				sb.append(month);
				sb.append("_");
				sb.append(j);
				sb.append(",id,cniStatus,0,100;");
			}
			for (int j = 1; j <= 6; j++) {
				sb.append("account_balance_");
				sb.append(year);
				sb.append(month);
				sb.append("_");
				sb.append(j);
				sb.append(",id,cniStatus,0,100;");
			}
			for (int j = 1; j < 7; j++) {
				sb.append("dofaofu_");
				sb.append(year);
				sb.append(month);
				sb.append("_");
				sb.append(j);
				sb.append(",id,cniStatus,0,100;");
			}
			sb.append("is_month_");
			sb.append(year);
			sb.append(month);
			sb.append(",id,cniStatus,0,100");
			if (i != 24) {
				sb.append(";");
			}
			if (month == 12) {
				month = 1;
				year++;
			} else {
				month++;
			}
		}
		System.out.println(sb.toString());
	}

	private Connection aConn = null;// 连接对象 a
	private Connection bConn = null;//oracle链接对象
	private Statement aStm = null;//sql server声明
	private Statement bStm = null;//oracle声明
	private ResultSet rs = null;//结果集合
	private List<TableInfo> tableInfos = new ArrayList<>();//表信息集合

	public SyncTask() {
		try {
			ConnInfo connInfo = new ConnInfo();
			aConn = connInfo.aConnCreate();
			if (aConn == null) {
				log.error("连接a初始化连接出错，请检查配置文件或网络！");
				return;
			}

			bConn = connInfo.bConnCreate();
			if (bConn == null) {
				log.error("连接b初始化连接出错，请检查配置文件或网络！");
				return;
			}

			aStm = aConn.createStatement();
			bStm = bConn.createStatement();
		} catch (Exception e) {
			log.error("new SyncTask异常", e);
		}
	}

	@Override
	public void run() {
		try {
			log.info("------定时任务开始--------");

			// 检查数据库连接是否正常，如果不正常则再次初始化
			if (!checkConnected(aConn)) {
				ConnInfo connInfo = new ConnInfo();
				aConn = connInfo.aConnCreate();
				if (aConn == null) {
					log.error("连接对象a初始化出错，请检查配置文件或网络！");
					return;
				}
			}

			if (!checkConnected(bConn)) {
				ConnInfo connInfo2 = new ConnInfo();
				bConn = connInfo2.bConnCreate();
				if (aConn == null) {
					log.error("连接对象b初始化出错，请检查配置文件或网络！");
					return;
				}
			}

			// 从配置文件读出要导的数据表信息
			if (initTablesInfo("o2s") == 1) {
				log.info("------------------开始从 oracle 导数据到 sql server -----------------");
				handleImport("o2s");
				log.info("------------------结束从 oracle 导数据到 sql Server -----------------\n");
			}

			if (initTablesInfo("s2o") == 1) {
				log.info("------------------ 开始从 sql Server 导数据到 oracle -----------------");
				handleImport("s2o");
				log.info("------------------ 结束从 sql server 导数据到 oracle -----------------\n");
			}

			if (initTablesInfo("o2o") == 1) {
				log.info("------------------ 开始从 oracle 导数据到 oracle -----------------");
				handleImport("o2o");
				log.info("------------------ 结束从 oracle导数据到 oracle -----------------\n");
			}

			if (initTablesInfo("m2m") == 1) {
				log.info("------------------ 开始从 mysql导数据到 mysql -----------------");
				handleImport("m2m");
				log.info("------------------ 结束从 mysql导数据到 mysql -----------------\n");
			}

			log.info("------定时任务结束--------\n\n");
		} catch (SQLException e) {
			e.printStackTrace();
		}
	}

	/**
	 * 从配置文件读出要导出的表配置数据
	 *
	 * @param importType
	 * @return
	 */
	public int initTablesInfo(String importType) {
		// 作为标识 1：查询到 0：未查询到
		int flag = 1;
		try {
			// 清理原变量里的数据
			tableInfos.clear();
			// 从配置文件读取基本配置
			String strTables = PropertiesUtils.getValue(importType);
			if (!strTables.equals("")) {
				String[] infos = strTables.split(";");
				if (infos.length > 0) {
					for (String info : infos) {
						String[] split = info.split(",");
						// 如果值没填够就会出错
						if (split.length != 5) {
							log.error("表配置信息出错，请按指定格式配置");
							flag = 0;
						} else {
							List<ColumnInfo> columnInfos = getColumnInfoList(split[0], importType);
							TableInfo tableInfo = new TableInfo(split[0], split[1], split[2], split[3], split[4], columnInfos);
							tableInfos.add(tableInfo);
						}
					}
				} else {
					log.error("表配置信息出错，请按指定格式配置");
					flag = 0;
				}
			} else {
				flag = 0;
			}
		} catch (Exception e) {
			log.error("initTablesInfo异常", e);
		}
		return flag;
	}

	/**
	 * 数据导入
	 *
	 * @param importType
	 */
	public void handleImport(String importType) {
		int count = 0;
		try {
			for (TableInfo tableInfo : tableInfos) {
				// 查询语句
				StringBuilder selectSql = new StringBuilder();
				// 插入语句的字段名
				StringBuilder inSqlKey = new StringBuilder();
				// 插入语句的字段值
				StringBuilder inSqlValue = new StringBuilder();
				selectSql.append("select ");
				int columnCount = tableInfo.getColumnInfoList().size();
				// 若字段为空，证明前面并未查到源库表，直接停止下面的逻辑运行
				if (columnCount == 0) {
					return;
				}
				for (int j = 0; j < columnCount; j++) {
					selectSql.append(tableInfo.getColumnInfoList().get(j).columnName);
					inSqlKey.append(tableInfo.getColumnInfoList().get(j).columnName);
					inSqlValue.append("?");
					if (j != columnCount - 1) {
						selectSql.append(",");
						inSqlKey.append(",");
						inSqlValue.append(",");
					}
				}
				selectSql.append(" from ").append(tableInfo.getTableName()).append(" where ").append(tableInfo.getSignField()).append("='").append(tableInfo.getBeforeVal()).append("'");
				log.info("查询源库未同步的数据：" + selectSql.toString());
				// 源库执行查询
				rs = aStm.executeQuery(selectSql.toString());

				// 插入数据
				PreparedStatement preparedStatement = null;
				StringBuilder insertSql = new StringBuilder();
				insertSql.append("insert into ").append(tableInfo.getTableName()).append("(");
				insertSql.append(inSqlKey);
				insertSql.append(") values(");
				insertSql.append(inSqlValue);
				insertSql.append(")");

				// 目标库缓冲连接初始化
				preparedStatement = bConn.prepareStatement(insertSql.toString());

				// 开始同步
				count = syncStart(columnCount, tableInfo, preparedStatement, insertSql, importType);

				rs.close();
				preparedStatement.close();
			}

			if (count > 0) {
				log.info("本次导入" + count + "条数据！");
			}
		} catch (Exception e) {
			log.error("导入数据失败!", e);
		}
	}

	/**
	 * 同步数据，并放回条数
	 * @param columnCount
	 * @param tableInfo
	 * @param preparedStatement
	 * @param insertSQL
	 * @param importType
	 * @return
	 * @throws SQLException
	 */
	private int syncStart(int columnCount, TableInfo tableInfo, PreparedStatement preparedStatement, StringBuilder insertSQL, String importType) throws SQLException {
		int count = 0;
		try {
			aConn.setAutoCommit(false);
			bConn.setAutoCommit(false);
			while (rs.next()) {
				// 注意，rs的set跟get是以1开始的，不是0
				for (int k = 1; k <= columnCount; k++) {
					switch (tableInfo.getColumnInfoList().get(k - 1).columnType) {
						// 修正数据类型
						case Types.VARCHAR:
						case Types.CHAR:
							String getVal = rs.getString(k);
							if (getVal != null && getVal.length() > 0) {
								preparedStatement.setString(k, rs.getString(k).trim());
							} else {
								preparedStatement.setString(k, "");
							}
							break;
						case Types.LONGVARBINARY:
						case Types.BIGINT:
							preparedStatement.setLong(k, rs.getLong(k));
							break;
						case Types.DATE:
							preparedStatement.setDate(k, rs.getDate(k));
							break;
						case Types.TIMESTAMP:
							preparedStatement.setTimestamp(k, rs.getTimestamp(k));
							break;
						case Types.SMALLINT:
							preparedStatement.setInt(k, rs.getInt(k));
							break;
						case Types.NUMERIC:
							switch (tableInfo.getColumnInfoList().get(k - 1).columnScale) {
								case 0:
									preparedStatement.setInt(k, rs.getInt(k));
									break;
								case -127:
									preparedStatement.setFloat(k, rs.getFloat(k));
									break;
								default:
									preparedStatement.setDouble(k, rs.getDouble(k));
							}
							break;
						default:
							preparedStatement.setString(k, rs.getString(k));
					}
				}

				log.info("数据插入目标库：" + insertSQL.toString());
				preparedStatement.executeUpdate();
				preparedStatement.executeBatch();

				// 修改同步数据的状态（源库声明）
				Statement updateStm = aConn.createStatement();

				StringBuilder updateSql = new StringBuilder();
				updateSql.append("update ").append(tableInfo.getTableName()).append(" ");
				updateSql.append("set ").append(tableInfo.getSignField()).append("='").append(tableInfo.getAfterVal()).append("' ");
				updateSql.append("where ").append(tableInfo.getKeyField()).append("='").append(rs.getString(tableInfo.getKeyField())).append("'");

				log.info("修改源库的状态值：" + updateSql);
				updateStm.execute(updateSql.toString());

				updateStm.close();
				count++;
			}
			aConn.commit();
			bConn.commit();
		} catch (Exception e) {
			// 出现异常就回滚
			aConn.rollback();
			bConn.rollback();
			log.error("commit抛出异常", e);
		}
		return count;
	}

	/**
	 * 获取字段信息（字段名、字段类型、字段长度）
	 * 确保目标库存在同名表，没有则创建
	 *
	 * @param tableName
	 * @param importType
	 * @return
	 */
	public List<ColumnInfo> getColumnInfoList(String tableName, String importType) {
		List<ColumnInfo> list = new ArrayList<>();
		try {
			// 逻辑：去源库找字段，因为源库首先必须存在表，并且可以根据源库来生成目标库
			// 源库查询语句
			String selectSqlFor = "select * from " + tableName;
			// 源库声明
			Statement statementFor = aConn.createStatement();
			// 目标库查询语句
			String selectSqlTo = "select * from " + tableName;
			// 目标库声明
			Statement statementTo = bConn.createStatement();
			ResultSet resultSet = Objects.requireNonNull(statementFor).executeQuery(selectSqlFor);
			ResultSetMetaData meta = resultSet.getMetaData();
			for (int i = 1; i <= meta.getColumnCount(); i++) {
				ColumnInfo columnInfo = new ColumnInfo();
				columnInfo.columnName = meta.getColumnName(i);
				columnInfo.columnType = meta.getColumnType(i);
				columnInfo.columnScale = meta.getScale(i);
				list.add(columnInfo);
			}
			// 检查目标库是否存在，若不存在则创建
			createTableIfNotExist(statementTo, selectSqlTo, tableName, meta, importType);
			resultSet.close();
			statementFor.close();
		} catch (Exception e) {
			log.error("---------------" + tableName + " 取字段名及类型时抛出异常（重点确认是否存在该源库）---------------");
		}
		return list;
	}

	/**
	 * 检查目标库是否存在，若不存在则创建
	 *
	 * @param statementTo
	 * @param selectSqlTo
	 * @param tableName
	 * @param meta
	 * @param importType
	 * @throws SQLException
	 */
	private void createTableIfNotExist(Statement statementTo, String selectSqlTo, String tableName, ResultSetMetaData meta, String importType) throws SQLException {
		// 需要求证目标库是否存在同名数据表
		// 若不存在 则需要根据字段和字段类型创建
		try {
			statementTo.executeQuery(selectSqlTo);
		} catch (Exception e) {
			StringBuilder createTableSql = new StringBuilder();
			createTableSql.append("create table " + tableName + "(");
			for (int i = 1; i <= meta.getColumnCount(); i++) {
				// 不同数据库类型的逻辑
				switch (importType) {
					case "o2s":
						createTableSql.append(meta.getColumnName(i) + " ");
						switch (meta.getColumnType(i)) {
							case Types.VARCHAR:
								createTableSql.append("varchar(200)");
								break;
							case Types.INTEGER:
							case Types.NUMERIC:
								createTableSql.append("int");
								break;
							default:
								createTableSql.append("varchar(200)");
								break;
						}
						break;
					case "s2o":
						createTableSql.append(meta.getColumnName(i)).append(" ");
						switch (meta.getColumnType(i)) {
							case Types.VARCHAR:
							case Types.NVARCHAR:
								createTableSql.append("varchar2(200)");
								break;
							case Types.INTEGER:
								createTableSql.append("number");
								break;
							default:
								break;
						}
						break;
					default:
						// 默认是同数据库类型情况
						createTableSql.append(meta.getColumnName(i)).append(" ");
						switch (meta.getColumnType(i)) {

						}
						break;
				}

				// 如果没到最后一个 加上逗号
				if (i != meta.getColumnCount()) {
					createTableSql.append(",");
				}
			}
			createTableSql.append(")");
			statementTo.executeUpdate(createTableSql.toString());
			log.info("==================" + tableName + "表创建完成==================");
		}
	}

	/**
	 * 检查数据库连接
	 *
	 * @param conn
	 * @return
	 * @throws SQLException
	 */
	private boolean checkConnected(Connection conn) throws SQLException {
		if (conn == null || conn.isClosed()) {
			return (false);
		}
		boolean connected = false;
		Statement stat = null;
		try {
			stat = conn.createStatement();
			stat.close();
			connected = true;
		} catch (SQLException e) {
			e.printStackTrace();
		} finally {
			try {
				if (stat != null) {
					stat.close();
				}
			} catch (SQLException e) {
				e.printStackTrace();
			}
		}
		return connected;
	}
}
